package com.boot2.core.conf.mqtt;

import com.boot2.core.mqtt.*;
import com.boot2.core.utils.MqttTopicUtil;
import lombok.extern.apachecommons.CommonsLog;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;

/**
 * @author zhangweilin
 * @date 2018/9/20 11:38
 * @description: mqtt收消息配置
 */
@Configuration
@Conditional(MqttInboundConditional.class)
@CommonsLog
public class MqttInboundConfiguration extends MqttConf {
    @Autowired(required = false)
    MqttMessageLocker mqttMessageLocker;
    @Autowired
    private MqttProperties mqttProperties;
    @Autowired(required = false)
    private List<MqttMessageRouter> mqttMessageRouterList;
    @Autowired(required = false)
    private MqttMessageInbound mqttMessageInbound;
    @Autowired(required = false)
    private MqttMessageEvent mqttMessageEvent;
    /**
     * 可拓展的转换器
     */
    @Qualifier("inoundPahoMessageConverter")
    @Autowired(required = false)
    private DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();

    @PostConstruct
    public void init() {
//        if (null != mqttMessageRouter) {
//            mqttMessageRouter.init();
//        }
        if (null != mqttMessageRouterList) {
            mqttMessageRouterList.forEach(e -> e.init());
        }
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() throws Exception {
        log.info("inbound.mqttProperties: " + mqttProperties);
        MqttInboundProperties inbound = mqttProperties.getInbound();
        List<String> sysTopicList = inbound.getSysTopics();
        List<String> busiTopicList = inbound.getBusiTopics();
        String lwtTopic = inbound.getLwtTopic();

        List<String> allTopicList = new ArrayList<>();
        List<String> allTopicList2 = new ArrayList<>();
        allTopicList.addAll(sysTopicList);
        allTopicList.addAll(busiTopicList);
        allTopicList.add(lwtTopic);
        System.out.println("allTopicList1 = " + allTopicList);
        //如果加了统一前缀，则追加
        String mqttTopicsPrefix = mqttProperties.getMqttTopicsPrefix();
        allTopicList.forEach(e -> {
            allTopicList2.add(mqttTopicsPrefix + e);
        });
        allTopicList.clear();
        allTopicList = null;
        System.out.println("allTopicList2 = " + allTopicList2);

        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId() + "_Inbound",
                mqttClientFactory(mqttProperties, inbound.getCleanSession()), allTopicList2.toArray(new String[]{}));
        adapter.setCompletionTimeout(inbound.getCompletionTimeout());
        adapter.setConverter(defaultPahoMessageConverter);
        adapter.setQos(inbound.getQos());
        adapter.setShouldTrack(inbound.getShouldTrack());
        adapter.setOutputChannel(mqttInputChannel());
        // adapter.addTopic(topic);
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
//                String id = message.getHeaders().getId().toString();
//                System.out.println("message.id = " + id);
//                String key = "mqttMessageId:" + id;

//                redisDao.setForSeconds(key, 1, 20L);
                if (null != mqttMessageLocker && !mqttMessageLocker.getDistLock("handleMessage_", message)) {
                    return;
                }

                //是否是业务消息,也即是否继续处理
                boolean flag = true;
                if (null != mqttMessageEvent) {
                    flag = mqttMessageEvent.handleMessage(message);
                } else {
                    log.error("mqttMessageEvent is null，未处理消息 message: " + message);
                }
                if (!flag) {
                    log.info("系统消息,无须作为业务处理");
                    return;
                }
                String mqttTopic = (String) message.getHeaders().get("mqtt_topic");
                //业务消息主题列表
                List<String> busiTopicList = mqttProperties.getInbound().getBusiTopics();
                if (CollectionUtils.isEmpty(busiTopicList)) {
                    return;
                }
                //当前主题是否是业务消息主题
                boolean hasCurrentTopic = false;
                for (int i = 0; i < busiTopicList.size(); i++) {
                    String topic = busiTopicList.get(i);
                    topic = MqttTopicUtil.getTopicPrefix(topic);
                    if (mqttTopic.startsWith(mqttProperties.getMqttTopicsPrefix() + topic)) {
                        hasCurrentTopic = true;
                        break;
                    }
                }
                if (!hasCurrentTopic) {
                    return;
                }
                //如果当前消息是完整的系统消息或遗愿消息，则不需要再当业务消息处理
                if (null != mqttMessageInbound) {
                    mqttMessageInbound.handleMessage(message);
                } else {
                    log.error("mqttMessageInbound is null，未处理消息 message: " + message);
                }

                //为防止集群中某个节点处理过快，在其它节点还没接收完消息情况下就已经处理完并释放了锁，导致其它节点还会重复处理，特在此，不主动释放
//                if (null != redisDistributedLock) {
//                    boolean releaseLock = redisDistributedLock.releaseLock(key, randomString);
//                    //如果对并发敏感，考虑此语句写在释放锁里的lua脚本里去
//                    redisDao.setForSeconds(key, 1, 5L);
//                }
            }
        };
    }
}
